package com.gitee.hermer.engine.jars.core.eventbus.factory.activemq;

import java.io.Serializable;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.apache.commons.collections.MultiHashMap;
import org.springframework.beans.factory.annotation.Autowired;
import com.gitee.hermer.boot.jee.commons.collection.StringCache;
import com.gitee.hermer.boot.jee.commons.exception.ErrorCode;
import com.gitee.hermer.boot.jee.commons.exception.PaiUException;
import com.gitee.hermer.boot.jee.commons.utils.StringUtils;
import com.gitee.hermer.boot.jee.commons.verify.Assert;
import com.gitee.hermer.engine.jars.core.app.domain.EngineAppMethodBean;
import com.gitee.hermer.engine.jars.core.domain.EngineConstant;
import com.gitee.hermer.engine.jars.core.eventbus.EventBusService;
import com.gitee.hermer.engine.jars.core.eventbus.EventType;
import com.gitee.hermer.engine.jars.core.eventbus.config.EventBusProperties;
import com.gitee.hermer.engine.jars.dependent.activemq.core.ActiveMQContext;
import com.gitee.hermer.engine.jars.dependent.activemq.core.pool.ActiveMQSessionPoolFactory;
import com.gitee.hermer.engine.jars.dependent.dsm.IEnginePropertiesAware;

public class ActiveMQEventBusService extends EventBusService<String, Message, ActiveMQMessageListener, ActiveMQExceptionListener,JMSException> implements IEnginePropertiesAware<EventBusProperties>{

	@Autowired
	private ActiveMQContext context;

	private EventBusProperties properties;



	protected Map<ClassLoader, MultiHashMap> consumers = new ConcurrentHashMap<ClassLoader, MultiHashMap>();


	// test junit 
	MultiHashMap  getConsumers(ClassLoader loader){
		return consumers.get(loader);
	}


	@Override
	public void sendQueueEventBus(String eventName, Serializable... args) throws JMSException{
		sendMessage(new ActiveMQQueue(getEventName(eventName)), args);
	}

	@Override
	public void sendBroadcastEventBus(String eventName, Serializable... args) throws JMSException{
		sendMessage(new ActiveMQTopic(getEventName(eventName)), args);
	}

	protected void sendMessage(Destination destination,Serializable... serializable) throws JMSException{
		MessageProducer producer = null;
		ActiveMQSessionPoolFactory factory = null;
		Session session = null;
		try{
			factory = context.getFactory(null);
			session = factory.getOneSession();
			Assert.notNull(session);
			producer = session.createProducer(destination);
			sendMessage(session,producer,serializable);
			if(context.isUseTransacted())
				session.commit();
		}catch (Exception e) {
			if(context.isUseTransacted() && session != null)
				session.rollback();
			throw e;
		}finally {
			if(producer != null)
				producer.close();
			if(factory != null && session != null){
				factory.returnSession(session);
			}else if(session != null){
				session.close();
			}
		}
	}

	private void sendMessage(Session session, MessageProducer producer, Serializable serializable) throws JMSException{
		ObjectMessage objectMessage = session.createObjectMessage(serializable);
		producer.send(objectMessage);
	}

	@Override
	public void register(String eventName, EngineAppMethodBean methodBean) {
		String eventType = (String)methodBean.getAnnotationAttributes().get(EngineConstant.ANNOTATE_FIELD_NAME_EVENT_TYPE);
		Assert.hasText(eventType);
		if(eventType.equals(EventType.BROADCAST)){
			this.topicMessageListener(eventName, new ActiveMQMessageListener(methodBean,context.isUseTransacted()));
		}else if(eventType.equals(EventType.QUEUE)){
			this.queueMessageListener(eventName, new ActiveMQMessageListener(methodBean,context.isUseTransacted()));
		}else{
			throw new PaiUException(ErrorCode.DATA_ERROR, StringUtils.format("事件类型错误！{}", eventType));
		}
	}

	@Override
	public boolean unregister(ClassLoader classLoader,String eventName) {
		try{
			if(consumers.containsKey(classLoader)){
				close(consumers.get(classLoader).getCollection(eventName));
				return consumers.get(classLoader).remove(eventName) != null;
			}
		}catch (Exception e) {
			error(e.getMessage(),e);
		}
		return false;
	}

	@Override
	public boolean unregisterAll(ClassLoader classLoader) {
		try{
			if(consumers.containsKey(classLoader)){
				close(consumers.get(classLoader).values());
				return consumers.remove(classLoader) != null;
			}
		}catch (Exception e) {
			error(e.getMessage(),e);
		}
		return false;
	}

	void close(Collection collection){
		if(collection != null){
			for (Object consumer: collection) {
				try{
					((MessageConsumer)consumer).close();
				}catch (Exception e) {
					error(e.getMessage(),e);
				}
			}
		}
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		List<ActiveMQSessionPoolFactory>  factorys = context.getAllFactory();
		for (ActiveMQSessionPoolFactory activeMQSessionPoolFactory : factorys) {
			activeMQSessionPoolFactory.getConnection().setExceptionListener(exceptionListener);
		}
	}

	@Override
	public void destroy() throws Exception {
		for (MultiHashMap hashMap : consumers.values()) {
			close(hashMap.values());
		}
	}

	@Override
	public void topicMessageListener(String event, ActiveMQMessageListener messageListener) {
		messageListener(event,new ActiveMQTopic(getEventName(event)), messageListener);
	}

	@Override
	public void queueMessageListener(String event, ActiveMQMessageListener messageListener) {
		messageListener(event,new ActiveMQQueue(getEventName(event)), messageListener);
	}


	public String getEventName(String event){
		return new StringCache(properties.getEventPrefix()).append(
				EngineConstant.STR_SPLIT).append(event).toString();
	}


	void putOf(ClassLoader classLoader,String key,MessageConsumer value){
		if(!consumers.containsKey(classLoader)){
			consumers.put(classLoader, new MultiHashMap());
		}
		consumers.get(classLoader).put(key, value);
	}

	protected void messageListener(String eventName,Destination destination,ActiveMQMessageListener messageListener){
		Assert.notEmpty(context.getAllFactory());
		List<ActiveMQSessionPoolFactory>  factorys = context.getAllFactory();
		for (ActiveMQSessionPoolFactory factory : factorys)
		{
			Session session = null;
			MessageConsumer consumer = null;
			try{
				session = factory.getOneSession();
				if(context.isUseTransacted()){
					if(destination instanceof ActiveMQTopic)
						messageListener.setTopicSession(session);
					else
						messageListener.setQueueSession(session);
				}
				consumer = session.createConsumer(destination);
				consumer.setMessageListener(messageListener);
				putOf(messageListener.getClassLoader(), eventName, consumer);
			}catch (Exception e) {
				if(session != null)
					factory.returnSession(session);
				exceptionListener.onError(e);
				if(consumer != null){
					try { consumer.close(); } catch (JMSException e1) { }
				}
			}
		}
	}

	@Override
	@Autowired
	public void setProperties(EventBusProperties properties) {
		this.properties = properties;
	}



}
